package cn.zwx.common.rocket.producer.service.impl;

import cn.zwx.base.enums.ResponseCodeEnum;
import cn.zwx.base.exception.BizWrapException;
import cn.zwx.common.rocket.producer.bo.MessageBO;
import cn.zwx.common.rocket.producer.bo.OrderMessageBO;
import cn.zwx.common.rocket.producer.service.DefaultMqProducerService;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Objects;

/**
 * @author zhangwenxue
 */
@Service
public class DefaultMqProducerServiceImpl implements DefaultMqProducerService {

    public static final Logger logger = LoggerFactory.getLogger(DefaultMqProducerServiceImpl.class);

    @Resource
    private DefaultMQProducer defaultMqProducer;
    @Override
    public void start() {
    }

    @Override
    public void shutDown() {
        if (Objects.nonNull(defaultMqProducer)){
            defaultMqProducer.shutdown();
        }
    }

    @Override
    public SendResult sendMsg(MessageBO bo) throws BizWrapException {
        Message message = new Message(bo.getTopic(), bo.getTag(), bo.getKey(), bo.getMessageBody().getBytes());
        if (Objects.nonNull(bo.getStartDeliverLevel())){
            message.setDelayTimeLevel(bo.getStartDeliverLevel());
        }
        SendResult sendResult;
        try {
            sendResult = defaultMqProducer.send(message);
        }catch (Exception e){
            logger.error("Send mq message failed. Topic is:{} Tag:{} key:{},body:{}",message.getTopic(), message.getTags(),message.getKeys(),
                    new String(message.getBody()),e);
            throw new BizWrapException(ResponseCodeEnum.SEND_MQ_ERROR);
        }
        logger.info("Send mq message success. Topic is:{} Tag:{} key:{} body:{}",message.getTopic(), message.getTags(),
                message.getKeys(),new String(message.getBody()));
        return sendResult;
    }

    @Override
    public void sendMsgForOneWay(MessageBO bo) throws BizWrapException {
        Message message = new Message(bo.getTopic(), bo.getTag(), bo.getKey(), bo.getMessageBody().getBytes());
        try {
            defaultMqProducer.sendOneway(message);
        }catch (Exception e){
            logger.error("Send mq message failed. Topic is:{} Tag:{} key:{},body:{}",message.getTopic(), message.getTags(),message.getKeys(),
                    new String(message.getBody()),e);
            throw new BizWrapException(ResponseCodeEnum.SEND_MQ_ERROR);
        }
        logger.info("Send mq message success. Topic is:{} Tag:{} key:{} body:{}",message.getTopic(), message.getTags(),
                message.getKeys(),new String(message.getBody()));
    }

    @Override
    public void sendMsgAsync(MessageBO bo, SendCallback callback) throws BizWrapException {
        Message message = new Message(bo.getTopic(), bo.getTag(), bo.getKey(), bo.getMessageBody().getBytes());
        try {
            defaultMqProducer.send(message,callback);
        }catch (Exception e){
            logger.error("Send mq message failed. Topic is:{} Tag:{} key:{},body:{}",message.getTopic(), message.getTags(),message.getKeys(),
                    new String(message.getBody()),e);
            throw new BizWrapException(ResponseCodeEnum.SEND_MQ_ERROR);
        }
        logger.info("Send mq message success. Topic is:{} Tag:{} key:{} body:{}",message.getTopic(), message.getTags(),
                message.getKeys(),new String(message.getBody()));
    }

    @Override
    public SendResult sendOrderMsg(OrderMessageBO bo, String shardingKey) throws BizWrapException {
        // 1.组装参数
        Message msg = new Message(
                // Message所属的Topic。
                bo.getTopic(),
                // Message Tag，可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在消息队列RocketMQ版的服务器过滤。
                bo.getTag(),
                // 设置代表消息的业务关键属性 请尽可能全局唯一 以方便您在无法正常收到消息情况下，可通过控制台查询消息并补发
                bo.getKey(),
                // Message Body，可以是任何二进制形式的数据，消息队列RocketMQ版不做任何干预，需要Producer与Consumer协商好一致的序列化和反序列化方式。
                bo.getMessageBody().getBytes()
        );
        msg.setKeys(bo.getOrderId());
        // 2.发送请求
        SendResult sendResult;
        try {
            sendResult =  defaultMqProducer.send(msg, (mqs, message, o) -> {
                long index = Long.parseLong(shardingKey) % mqs.size();
                return mqs.get((int) index);
            },shardingKey);
        }catch (Exception e){
            logger.error("Send mq message failed. Topic is:{} Tag:{} key:{},body:{}",msg.getTopic(), msg.getTags(),msg.getKeys(),
                    new String(msg.getBody()),e);
            throw new BizWrapException(ResponseCodeEnum.SEND_MQ_ERROR);
        }
        logger.info("Send mq message success. Topic is:{} Tag:{} key:{} body:{}",msg.getTopic(), msg.getTags(),
                msg.getKeys(),new String(msg.getBody()));
        return sendResult;
    }


}
